use std::{sync::{mpsc::{sync_channel, Receiver, SyncSender}, Arc, Mutex}, task::Context, time::Duration};
use futures::{future::BoxFuture, task::{waker_ref, ArcWake}};
use futures::FutureExt; // 引入 FutureExt 以使用 .boxed()
use rs_timer_future::TimerFuture;

fn main() {
    println!("future executor!!");
    println!("初始化执行器和生成器");
    let (executer, spawner) = new_executor_and_spawner();

    // 生成一个任务
    // 当我们使用 async 定义一个代码块的时候，实际上我们是在定义一个可以返回 Future 的函数或表达式，
    // 这个Future可以在稍后被轮询，通过调用 .poll 方法，以推进它的执行状态
    spawner.spawn(async {
        // 在 async 内部，可以用 .await 来暂停当前Future任务的执行，直到该Future完成，而不会阻塞整个线程。
        println!("start spawn async...");
        // 创建计时器Future，并异步等待它完成
        TimerFuture::new(Duration::new(2, 0)).await;
        println!("done...");
    });

    // 经过 spawner.spawn 之后，future就被以封装好的任务的形式抛进了 通道之中，等待执行器消费
    // 此时此刻生成器已经失去了意义，我们将其drop 掉，这样的话，channel的发送端就关闭了，接收端就会收到Err而跳出阻塞和循环
    drop(spawner);

    // 执行
    executer.run();
}


/// 执行器执行包含Future的Task任务
/// 1. 用户调用Spawner::spawn 方法传入一个满足Future特征的对象。
/// 2. spawn方法将Future包装成一个Task，并通过task_sender发送到任务通道中
/// 3. 任务随后会被Executor接受并执行
/// /// Executor 持有任务通道的接收端（Receiver），Spawner 持有发送端（SyncSender）
/// /// Receiver和SyncSender是Rust标准库中用于实现同步通道sync channel的类型，他们分别代表了通道的接收端和发送端
/// /// 这两个类型的变量通过 std::sync::mpsc::sync_channel 函数创建
/// 
/// /// 如下执行器和生成器中的Receiver和SyncSender是匹配的一对，一个负责发送，一个负责接收。

// task start +++++++++++++++++

/// 任务
/// 该任务携带着一个future,并且可以将自己调度放入任务通道之中，然后等待执行器去 poll
/// task_sender用于将自身重新放回任务通道中，以便执行器可以继续对其调用poll。
/// BoxFuture是一个future的封装
/// type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
/// + Pin<Box... 表示使用Pin和Box来确保指针指向的数据在其生命周期内不会被移动，这对于需要稳定内存地址的操作（如异步任务) 是必须的
/// + dyn Future<Output=T> 说明是一个动态分发的Future,这意味着它可以指向任何实现了Future特征的具体类型
/// + Send 表示该Future都可以在线程之间安全传递，并发会用到。
/// + 'a 是生命周期，表示 Future 的存活时间范围。
struct Task {
    future: Mutex<Option<BoxFuture<'static, ()>>>,
    task_sender: SyncSender<Arc<Task>>
}

// 执行器poll一个Future之前需要先调用wake方法进行唤醒
// 然后再由Waker负责调度该任务并将其放入任务通道中
// 创建Waker最简单地方式就是实现ArcWake特征
impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        // Task wake 地方式就是 将任务发送到任务通道中去，这样wake之后，任务就能被执行器poll了
        // 因为任务通道地接收端被执行器持有者，这就相当于通过wake告知了执行器可以继续执行 future 了
        // 参数 arc_self 是一个 Arc 管理的对象，也就是线程安全的引用计数只能指针。
        //      我们可以通过 clone 方法来增加引用计数指向相同的内存对象。
        let arc_self_clone = arc_self.clone();
        arc_self.task_sender.send(arc_self_clone).expect("任务队列已满!!");
    }
}
// 实现了 ArcWake 的Task就具备了wake executor的能力了

// task end ------------------------

// execute start +++++++++++++++++++++

/// 任务执行器
/// 任务的消费之：负责从通道中接受任务，然后执行
/// Executor不断从ready_queue中接受任务。对每个任务调用其poll方法，尝试驱动其完成
struct Executor {
    ready_queue: Receiver<Arc<Task>>
}

// executor持有着通道的接收端，如果我们想让他真正的执行运行能力，还得run起来，去消费
// 核心逻辑就是：从通道接收端拿到任务，然后
impl Executor {
    fn run(&self) {
        // 从通道中读取内容，阻塞着读
        while let Ok(task) = self.ready_queue.recv() {
            // 此时就相当于解除阻塞得到了一个task
            let mut futurebox = task.future.lock().unwrap();
            // futurebox 是一个 Option 其中包裹着 FutureBox，FutureBox中包裹着 future
            // 将这个Option取出来，之前的干掉，如果有的话。
            if let Some(mut futurereal) = futurebox.take() {
                // futurereal是 BoxFuture<T>
                // task实现了ArcWaker 可以使用waker_ref方法创建一个标准的 Waker
                // waker_ref是fetures包提供的一个函数，用于从实现了特定trait的类型中创建一个Waker
                let wakereal = waker_ref(&task);
                // 创建一个context，包含Waker
                let mut ctx = Context::from_waker(&*wakereal);
                // 开始操作future
                // 使用 as_mut() 方法脱掉 Box 得到可变引用 Pin<&mut dyn Future + Send + 'static>
                // 执行poll，poll操作尝试执行任务，如果执行完毕就 is_ready，如果没得执行就继续 is_pending
                if futurereal.as_mut().poll(&mut ctx).is_pending() {
                    // 还没准备好的话，将future重新放回去
                    // 刚才把 futurebox 给take走了，现在要重新把 futurebox 恢复了
                    // futurebox是 MutexGuard ，也就是被 Mutex 包裹的一个可变变量，我们可以直接修改它
                    *futurebox = Some(futurereal)
                }

            }
        }
    }
}


// execute end ------------------------

// spawner start +++++++++++++++++++++++

/// Spawner 负责创建新的 Future 然后将他发送到 任务通道中
/// 任务的生产者
/// 将用户提供的Future包装成一个Task，并通过task_sender发送到任务通道中
#[derive(Debug)]
struct Spawner {
    task_sender: SyncSender<Arc<Task>>
}

// 为 Spawner 添加一个方法用于生成 Future, 然后将它放入任务通道中
impl Spawner {
    // Send 约束确保该 Future 可以在线程之间安全地传递
    // 'static 表示该 Future 没有生命周期限制（即它可以被安全地存储和传递）
    fn spawn(&self, future: impl Future<Output = ()> + 'static + Send) {
        // future.boxed() 将 future 转换为 BoxFuture<'static, ()>。
        // 引入 FutureExt 以使用 .boxed()
        let future = future.boxed();
        let task = Arc::new(Task{
            future: Mutex::new(Some(future)), // 
            task_sender: self.task_sender.clone(), // 因为通道实现的是 多发送者/单接收者，所以不断地克隆发送者
        });

        // 将任务发送到队列之中
        self.task_sender.send(task).expect("任务队列已满!!");
    }
}


// spawner end ------------------------


// 使用该方法产出一对基于通道的消息发送者和消息接收者
// 消息发送者 构造 Executor 变量
// 消息接收者 构造 Spawner 变量
fn new_executor_and_spawner() -> (Executor, Spawner) {
    const MAX_QUEUED_TASKS: usize = 10_000;
    // 创建一个缓冲区为 10 的同步通道，task_sender可以写入数据，ready_queue 可以读取数据。
    // task_sender调用send方法发送
    // ready_queue调用recv方法读取
    let(task_sender, ready_queue) = sync_channel(MAX_QUEUED_TASKS);
    (Executor{ready_queue}, Spawner{task_sender})
}

